package com.aotain.nyx.abnormal;

import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.math3.distribution.NormalDistribution;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.apollo.DosSliding;
import com.aotain.apollo.SlidingWindow;
import com.aotain.common.CommonDB;
import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.statis.AbnStatisTuple;

public class AbnormalCheckImport implements FlatMapFunction<Tuple2<String,AbnormalStatTuple>, AbnStatisTuple> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = -6636160655852119780L;

	private HashMap<String,Double> dipPInfo = new HashMap<String,Double>();//各个目标Ip的熵值
	private HashMap<String,Long> dipPVMap = new HashMap<String,Long>();//目标IP PV数
	private HashMap<String,Long> sdPVMap = new HashMap<String,Long>();//各个源ip到目标ip的PV数
	private Map<String,Long> tripleMap = new HashMap<String,Long>();//各个元组的PV
	private Map<String,Set<String>> dipToTriple = new HashMap<String,Set<String>>();//目标ip下的三元组映射
	private Map<String,Long> delaytimeMap = new HashMap<String,Long>();//求延时Map
	private Map<String,Long> maxDelayMap = new HashMap<String,Long>();//最大延时Map
	private Map<String,Long> minDelayMap = new HashMap<String,Long>();//最小延时Map
	private Map<String,Long> streamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> streamOctetsMap = new HashMap<String,Long>();//
	private Map<String,Integer> delayTimePVMap = new HashMap<String,Integer>();//
	private HashMap<String,Set<String>> rowKeyMap = new HashMap<String,Set<String>>();//
	private HashMap<String,Long> abnStartTime = new HashMap<String,Long>(); //记录异常开始（第1分钟）
	private HashMap<String,Long> abnUpdateTime = new HashMap<String,Long>(); //DIP 异常追踪时间
	
	//DIP异常状态，1：发现异常等待持续跟踪，2：确认异常状态 
	private HashMap<String,Integer> abnDIPStatus = new HashMap<String,Integer>();
	
	//DIP+SIP 入库更新时间
	private HashMap<String,Long> abnRowKeyUpdate = new HashMap<String,Long>(); 
	
	//一个连接的异常开始时间
	private HashMap<String,String> abnRowKeyStart = new HashMap<String,String>(); 
	
	//当发现异常，并且启动入库状态是，将当前区间的流量做完成的统计
	private Map<String,Long> memStreamPacketMap = new HashMap<String,Long>();//
	private Map<String,Long> memStreamOctetsMap = new HashMap<String,Long>();//

	String dbJson = null;
	private SlidingWindow cache = new SlidingWindow(20);
	private DosSliding dipPvCache = new DosSliding(20);
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();

	@Override
	public void flatMap(Tuple2<String, AbnormalStatTuple> tuple,
			Collector<AbnStatisTuple> out) throws Exception {
		dipPVMap = tuple.f1.getDipPVMap();
		sdPVMap = tuple.f1.getSdPVMap();
		tripleMap = tuple.f1.getTripleMap();
		dipToTriple = tuple.f1.getDipToTriple();
		delaytimeMap = tuple.f1.getDelaytimeMap();
		maxDelayMap = tuple.f1.getMaxDelayMap();
		minDelayMap = tuple.f1.getMinDelayMap();
		streamPacketMap = tuple.f1.getStreamPacketMap();
		streamOctetsMap = tuple.f1.getStreamOctetsMap();
		delayTimePVMap = tuple.f1.getDelayTimePVMap();
		
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_m1  = new SimpleDateFormat("yyyyMMddHHmm00");
		SimpleDateFormat sdf_m2  = new SimpleDateFormat("yyyy-MM-dd HH:mm:00");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
//		sdpvCache.addPInfo(sdPVMap);
		dipPvCache.addPInfo(dipPVMap);
		//求熵
		for(Entry<String,Set<String>> entry:dipToTriple.entrySet()) {
			String dip_idc_dp = entry.getKey();
			Set<String> tripleSet = entry.getValue();
			Long dipPV = dipPVMap.get(dip_idc_dp);
			if(dipPV == 0) {
				//如果概率为0，说明值为0，不计算信息熵，直接返回
				continue;
			}
			double pInfo = 0.0;
			for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
				String triple = iter.next();
				
				if(tripleMap.get(triple) == 0) {
					continue;
				}
				
				Double p = (double)tripleMap.get(triple)/(double)dipPV;
				
				if(p.isNaN()) {
					Logger.getLogger(AbnormalCheckImport.class)
						.info(String.format("###ABN LOG, Check tripleMap is NaN,key:%s value:%d dipPV:%d", triple,tripleMap.get(triple),dipPV));
				}
				//如果p=0 计算信息熵的方法就无法执行 结果为NaN
				Double tmpInfo = -Math.log(p)*p;
				pInfo += tmpInfo;
			}
			if(pInfo != 0.0) {
				dipPInfo.put(dip_idc_dp, pInfo);
			} else {
				//如果有一个阶段信息熵为0，清除窗口
				cache.clearZeroP(dip_idc_dp);
			}
		}

		cache.addPInfo(dipPInfo);
		Map<Integer,HashMap<String,Double>> infoMap = cache.getTuplesWindow();//得到滑动窗口中关于熵的数据
		Map<String,Integer> pInfoLengthMap = new HashMap<String,Integer>();//求熵值平均值时的长度
		Map<String,Double> totalP = new HashMap<String,Double>();//当前所有窗口对应目标元组的熵值和的Map
		
		if(infoMap != null ) {
			for(HashMap<String,Double> singleMap: infoMap.values()) {
				if(singleMap != null) {
					for(Entry<String,Double> entry : singleMap.entrySet()) {
						String dip_idc_dp = entry.getKey();
						Double value = entry.getValue(); //信息熵
						Double totalPvalue = 0.0;
						if(totalP.containsKey(dip_idc_dp)) {
							totalPvalue = totalP.get(dip_idc_dp);
						}
						
						//						if(value != 0) {
						Integer sum = pInfoLengthMap.get(dip_idc_dp);
						if(sum == null) {
							sum = 0;
						}
						sum ++;
						pInfoLengthMap.put(dip_idc_dp, sum);
						if(totalPvalue == null || totalPvalue.isNaN()) {
							totalPvalue = value;
						} else {
							totalPvalue += value;
						}
						totalP.put(dip_idc_dp, totalPvalue);
					}
				}
			}

		}

		Map<String,Double> avgP = new HashMap<String,Double>();//目标元组熵值的均值
		for(String dkey : totalP.keySet()) {
			Double avgPvalue = totalP.get(dkey) / (double) pInfoLengthMap.get(dkey);
			if(avgPvalue.isNaN()) {
				Logger.getLogger(AbnormalCheckImport.class)
					.info(String.format("###ABN LOG, Check avgPvalue totalP [%f],pInfoLengthMap:[%d],Key:%s",
							totalP.get(dkey),pInfoLengthMap.get(dkey),dkey));
			}
			avgP.put(dkey, avgPvalue);
		}

		HashMap<String,Double> subsqrMap = new HashMap<String,Double>();
		for(String key : avgP.keySet()) {
			Double value = 0.0;
			if(dipPInfo.get(key) != null && avgP.get(key)!= null) {
				value = (dipPInfo.get(key)-avgP.get(key))*(dipPInfo.get(key)-avgP.get(key));
				if(value.isNaN()) {
					Logger.getLogger(AbnormalCheckImport.class)
						.info(String.format("###ABN LOG, Check Pinfo[%f],avgP:[%f],Key:%s",dipPInfo.get(key),avgP.get(key),key));
				}
			}
			subsqrMap.put(key, value);
		}
		cache.addSubsqr(subsqrMap);

		Map<String,Double> totalSubsqr = new HashMap<String,Double>();//差的平方和汇总
//		Map<String,Double> stdMap = new HashMap<String,Double>();//标准差集合
		Map<Integer,HashMap<String,Double>> winSums = cache.getSumWindow();
//		Map<Integer,HashMap<String,Long>> abnNumMaps = sdpvCache.getTuplesWindow();


		for(HashMap<String,Double> singleMap: winSums.values()) {
			if(singleMap != null) {
				for(Entry<String,Double> entry : singleMap.entrySet()) {
					String key = entry.getKey();
					Double value = entry.getValue();
					Double totalSvalue = 0.0;
					
					if(totalSubsqr.containsKey(key)) {
						totalSvalue = totalSubsqr.get(key);
					}
					
					if(value.isNaN()) {
						value = 0.0;//后续计算都出现NaN的情况，检测是否由于value = NaN造成。
					}
					
					if(totalSvalue==null || totalSvalue.isNaN()) {
						totalSvalue = 0.0;
					} else {
						totalSvalue += value;
					}
					totalSubsqr.put(key, totalSvalue);
				}
			}
		}

		
		
		Date date = new Date();
		Timestamp dateStr_m = new Timestamp(sdf_m2.parse(sdf_m2.format(date)).getTime());
		String dateStr = sdf.format(date);
		//*********Modified 2017-06-20
		//连续三分钟是否都在置信区间内
		for(String key : totalSubsqr.keySet()) {
			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) > 15) {//至少满足连续累积数值超过15个时间窗再做异常判断

				if(totalSubsqr.get(key).isNaN()) {//Turk 2017-07-10
					Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG, totalSubsqr is NaN. Key:" + key);
					continue;
				}
				
				double stdValue = 0.0;
				double sxn = 0.0;
				double confidenceValueMax = 0.0;//置信区间上界
				double confidenceValueMin = 0.0;//置信区间下界
				double nordis = 0.0;
				double avg = 0.0;
				double dipP = 0.0;
				NormalDistribution s = new NormalDistribution (0,1);//求正态分布分位数所用
				
				stdValue = Math.sqrt(totalSubsqr.get(key)/((double)pInfoLengthMap.get(key)-(double)1));//S_n
				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
				nordis = s.cumulativeProbability(sxn);
				avg = (avgP.get(key)==null || avgP.get(key).isNaN())?0.0:avgP.get(key);
				dipP = (dipPInfo.get(key)==null||dipPInfo.get(key).isNaN())?0.0:dipPInfo.get(key);
				confidenceValueMax = avg + nordis;
				confidenceValueMin = avg - nordis;

				if(dipP >=confidenceValueMin && dipP <= confidenceValueMax)
				{
					//如果存在正在检测异常的DIP，这个时候将状态设置为0，表示当前分钟没有异常。
					if(abnDIPStatus.containsKey(key)) {
						abnDIPStatus.put(key,0);
						//Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG, 中断,状态设置为0.KEY:" + key);
					}
				} else {//超出置信区间，开始记录异常行为
					
					//##########Turk 2017-07-10 #########
					if(!abnStartTime.containsKey(key) || abnDIPStatus.get(key) == 0) {
						//第一次发现,或者中断后又发生
						abnStartTime.put(key,System.currentTimeMillis());
						//更新状态
						abnDIPStatus.put(key, 1);
						abnUpdateTime.put(key, System.currentTimeMillis());
					} else {
						//已记录发现，判断是否持续
						if(abnDIPStatus.get(key) != 2 
								&& System.currentTimeMillis() - abnStartTime.get(key) >= 3*60*1000 - 5*1000L) {
							//不在入库状态，并且开始时间距离当前时间大于2分55秒，说明行为持续，开始入库异常日志
							abnDIPStatus.put(key, 2);
							
							Logger.getLogger(AbnormalCheckImport.class)
								.info(String.format("###ABN LOG, DIP[%s] 检测到持续异常.[%f-%f] {%f}"
										+ "[stdValue:%f sxn:%f nordis:%f avg:%f] {CK:%f/%d}", key,
										confidenceValueMin,confidenceValueMax,dipP,stdValue,sxn,nordis,avg,
										totalSubsqr.get(key),pInfoLengthMap.get(key)));
							
						} else {
							//未到达门限时间，继续等待
						}
						
						if(abnDIPStatus.get(key) == 2) { //更新异常持续时间
							abnUpdateTime.put(key, System.currentTimeMillis());
						}
						
					}
				}
			} 
		}

		//公式是：S_n=√(1/(n-1) ∑_(i=1)^n?〖(x_i-x)〗^2 ) 、S_xn=S_n/√n
		for(String key : totalSubsqr.keySet()) {

			String dip = key.split("\\*",-1)[0];
			String idcno = key.split("\\*",-1)[1];
			String dport = key.split("\\*",-1)[2];

			if(pInfoLengthMap.get(key) != null && pInfoLengthMap.get(key) >1) {
				//				Date date = new Date();
//				stdValue = Math.sqrt(totalSubsqr.get(key)/(pInfoLengthMap.get(key)-1));//S_n
//				sxn = stdValue / Math.sqrt(pInfoLengthMap.get(key));//S_xn=S_n/√n
//				nordis = s.cumulativeProbability(sxn);
//				avg = avgP.get(key)==null?0.0:avgP.get(key);
//				dipP = dipPInfo.get(key)==null?0.0:dipPInfo.get(key);
//				confidenceValueMax = avg + nordis; 	
//				confidenceValueMin = avg - nordis;

				Set<String> tripleSet = dipToTriple.get(key);//DIP中当前分钟的所有SIP集合

				if(tripleSet != null) {
					
					
					/* Turk 2017-07-10
					 * 1、DIP 发现持续异常，开启允许入库的窗口，在这个窗口内的SIP 都记录入库
					 * 2、新发现的SIP，在开始或者持续中，都以这个SIP第一次出现的那一分钟作为开始时间
					 * 3、在判定DIP 异常结束的时候，将这个阶段入库的SIP 全都设置为结束
					 * 4、在DIP 异常持续中，如果SIP 未持续第二分钟，则在当前分钟结束
					 */
					//$$$$Modified by turk 2016-01-04 ----END

					Set<String> rowKeySet = rowKeyMap.get(key)==null?new HashSet<String>():rowKeyMap.get(key);
					boolean check = true;//信息熵入库校验所需 Modified 2017-06-06
					for(Iterator<String> iter=tripleSet.iterator();iter.hasNext();) {
						String triple = "";
						String gis = "";
						try
						{
							boolean isNormal = true;
							triple = iter.next();
							long packets = streamPacketMap.get(triple);
							String score = Score(packets);
							
							String[] splits = triple.split("\\*",-1);
							String destport = splits[1];
							gis = splits[3];
							String protoType = splits[4];
							String sip = splits[2];
							String sdIP = dip + "*" + idcno + "*" + sip;
	
							String[] gisArray = gis.split("#",-1);
							String destAreaName = gisArray[0];
							String destGis = gisArray[1];
							String sourceAreaName = gisArray[2];//市
							String sourceGis = gisArray[3];
							String sourceAreaCountry = gisArray[4];
							String sourceAreaId = gisArray[5];
							String areaName = gisArray[6];//省
							String sourceAreaCityId = gisArray[7];
							String sourceAreaProvinceId = gisArray[8];
							String evaluate = "";
							String desc = "";
							String abnormal = "0";
							String delayTime = "";
							String rowkey_ = "";
							long attnum = 0;
							String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
							abnormal = "1";
							delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
							attnum = sdPVMap.get(sdIP);
	//						System.out.println("#######1 key=" + sdIP + "; time: " + delaytimeMap.get(sdIP) + ";pv: " + delayTimePVMap.get(sdIP) + "; delayTime" + delayTime);
							
							
							
							String accessTimeKey = dip + "*" + sip + "*" + destport;
							String rowKeyAbnormal = dip + "_" + dateStr + "_" + sip + "_" + destport + "_" + 1;
							if(abnDIPStatus.get(key) != null) {
								//记录攻击开始时间，由于入库的时候是以源IP为粒度入库，所以开始时间记录为当前时间							
								if(abnRowKeyStart.containsKey(accessTimeKey)) {
									dateStr = abnRowKeyStart.get(accessTimeKey);
								}
								
								if(abnDIPStatus.get(key) == 2 && abnRowKeyUpdate.get(rowKeyAbnormal) == null) {
									//DIP为确认异常状态，并且当前ROWKEY 未入库时
									Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG IMPORT,KEY=" + key + " ROWKEY= " + rowKeyAbnormal);
									abnRowKeyUpdate.put(rowKeyAbnormal, System.currentTimeMillis()); //INSERT
									abnRowKeyStart.put(accessTimeKey, dateStr);
									InsertLog(rowKeyAbnormal,triple,score,out);
									isNormal = false;
								} else if (abnRowKeyUpdate.get(rowKeyAbnormal) != null){
									//当前ROWKEY 已入库，持续更新记录
									Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG UPDATE,KEY=" + key + " ROWKEY= " + rowKeyAbnormal);
									abnRowKeyUpdate.put(rowKeyAbnormal, System.currentTimeMillis()); //UPDATE
									UpdateLog(rowKeyAbnormal,triple,out);
									isNormal = false;
								} else if (abnDIPStatus.get(key) == 0) {
									//异常消除，判断是否超过3分钟，满足条件则结束异常告警
									if(abnRowKeyUpdate.get(rowKeyAbnormal) != null && System.currentTimeMillis() - abnRowKeyUpdate.get(rowKeyAbnormal) >= 3*60*1000L) {
										Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG END,KEY=" + key 
												+ "END TIME:" + abnRowKeyUpdate.get(rowKeyAbnormal) + "ROWKEY:" + rowKeyAbnormal);
										UpdateLog(rowKeyAbnormal,triple,out);
										EndLog(rowKeyAbnormal);
										abnRowKeyUpdate.remove(rowKeyAbnormal);
										abnRowKeyStart.remove(accessTimeKey);
										mapHighFlow.remove(rowKeyAbnormal);
									}
									isNormal = false;
								}
							} else {
								//DIP 异常以停止，终止其他正在告警的SIP（等待3分钟）
								if(abnRowKeyUpdate.get(rowKeyAbnormal) != null && System.currentTimeMillis() - abnRowKeyUpdate.get(rowKeyAbnormal) >= 3*60*1000L) {
									Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG END,KEY=" + key 
											+ "END TIME:" + abnRowKeyUpdate.get(rowKeyAbnormal) + "ROWKEY:" + rowKeyAbnormal);
									UpdateLog(rowKeyAbnormal,triple,out);
									EndLog(rowKeyAbnormal);
									abnRowKeyUpdate.remove(rowKeyAbnormal);
									abnRowKeyStart.remove(accessTimeKey);
									mapHighFlow.remove(rowKeyAbnormal);
								}
							}
							
							
							if(score.equals("100")) {
								isNormal = true;
							}
							
							
							if(isNormal) {
								//正常日志入库SDS_IDC_LOG     
								//							Date date2 = new Date();
								//							String dateStr2 = sdf.format(date2);	
								//							rowKeyNormal = dip + "_" + dateStr2 + "_" + sip + "_" + dport + "_" + 0;	
								rowkey_ = dip + "_" + dateStr + "_" + sip + "_" + destport + "_" + 0;	
								evaluate = "100";
								abnormal = "0";
								desc = "连接";
								attnum = sdPVMap.get(sdIP);
								delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
								InsertNormalLog(rowkey_,triple);
	//							collector.emit("ABNORMAL_LOG", new Values(rowkey_,dip,sip,destport,dateStr,sourceAreaName,sourceGis,sourceAreaCountry,destAreaName,destGis,evaluate,abnormal,desc,attnum+"",delayTime ,"0"));
							}
						
						}catch(Exception ex) {
							Logger.getLogger(AbnormalCheckImport.class).info( 
									"ROW" + triple + "GIS Format:" + gis + " TRIPLESET:" + tripleSet.toString(), ex);
							throw ex;
						}

					}
					
					
					
					if(abnUpdateTime.get(key) != null && System.currentTimeMillis() - abnUpdateTime.get(key) >= 3*60*1000L) {
						Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG END,CLEAN MAP,KEY=" + key 
								+ "END TIME:" + abnUpdateTime.get(key));
						abnUpdateTime.remove(key);
						
						if (abnDIPStatus.get(key) !=null && abnDIPStatus.get(key) == 0) {
							abnDIPStatus.remove(key);
							abnStartTime.remove(key);
						}
					}
					
					
					
					
					if(rowKeySet.size()>0) {
						rowKeyMap.put(key, rowKeySet);
					}
					/**
					 * 辅助：近20分钟信息熵入库
					 */
					if(!check) {
						List<String> sqlList = new ArrayList<String>();
						for(Entry<Integer,HashMap<String,Double>>entryP : infoMap.entrySet()) {
							int i = entryP.getKey();
							HashMap<String,Double> infoPMap = entryP.getValue();
							if(infoPMap.containsKey(key)) {
								double dp = infoPMap.get(key);
								String sql = "insert into SDS_INFOP(DIP,REPORTTIME,NUM,PINFO,DPORT) "
										+ "	values('"+ dip +"','" + dateStr_m + "',"+ i +","+dp+","+ dport +" )";
								sqlList.add(sql);
								//									System.out.println("########### PINFO test sql is "+ sql);
							}

						}
						try {
							CommonDB.executeBatchJson(sqlList, dbJson);
						} catch (SQLException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
							Logger.getLogger("##############AbnormalCalBolt PInfo into oracle Excepetion : "+ e.getMessage());
						}
					}
				}
			}
			
			//#######TURK 2017-07-10##########
			ArrayList<String> removeKey = new ArrayList<String>();
			for(Entry<String,Long> entry : abnRowKeyUpdate.entrySet()) {
				//检查是否有ROWKEY 需要结束状态,如果超过3分钟，存在SIP没有数据更新，则认为结束，开启结束流程
				if(System.currentTimeMillis() - entry.getValue() >= 3*60*1000L) {
					Logger.getLogger(AbnormalCheckImport.class).info("###ABN LOG END,KEY=" + key 
							+ "END TIME:" + abnRowKeyUpdate.get(key) + "ROWKEY:" + entry.getKey());
					removeKey.add(entry.getKey());
					//String rowKeyAbnormal = dip + "_" + dateStr + "_" + sip + "_" + destport + "_" + 1;
					dip = entry.getKey().split("_",-1)[0];
					String sip = entry.getKey().split("_",-1)[2];
					String destport = entry.getKey().split("_",-1)[3];
					String accessTimeKey = dip + "*" + sip + "*" + destport;
					String rowKeyAbnormal = entry.getKey();
					EndLog(rowKeyAbnormal);
					abnRowKeyStart.remove(accessTimeKey);
				}
			}
			
			for(String remove : removeKey) {
				abnRowKeyUpdate.remove(remove);
			}
			//*******TURK 2017-07-10 *******
			
//			stdMap.put(key, stdValue);
		}

		tripleMap.clear();//各元组pv清零
		dipPVMap.clear();//dipPV清零
		dipPInfo.clear();
		dipToTriple.clear();
		sdPVMap.clear();
		delaytimeMap.clear();
		maxDelayMap.clear();
		minDelayMap.clear();
		delayTimePVMap.clear();
		streamPacketMap.clear();
		streamOctetsMap.clear();
		cache.advanceHead();
//		sdpvCache.advanceHead();
		dipPvCache.advanceHead();

	}
	
	
	private void InsertNormalLog(String rowKeyNormal,String rowValue){
		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String flowDir = splits[6];
		String sdIP = dip + "*" + idcno + "*" + sip;
		
		String dateStr = rowKeyNormal.split("_",-1)[1];

		String[] gisArray = gis.split("#",-1);
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];//市
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];
		String areaName = gisArray[6];//省
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		String desc = "正常连接";
		String delayTime = "";
		
		long attnum = 0;
		String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
		delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
		attnum = sdPVMap.get(sdIP);
		
		
		String sAbnormalTable = "SDS_ABNORMAL_LOG";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("ROWKEY", rowKeyNormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", destport));
		row.add(new DataColumn("ACCESSTIME", Long.parseLong(dateStr)));
		row.add(new DataColumn("ABRNORMAL", 0));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", attnum+""));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "100"));
		row.add(new DataColumn("PROVINCE", areaName));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));
		row.add(new DataColumn("UPSTREAMOCTETS", streamOctetsMap.get(triple)));
		row.add(new DataColumn("UPSTREAMPACKET", streamPacketMap.get(triple)));
		
//		System.out.println("NormalLog #######Into MongoDB#####" + row.toString());
		importtool.InsertRowData(sAbnormalTable, row);	
		
	}
	
	/**
	 * 新发现的异常连接
	        * @param rowKeyAbnormal
	        * @param row
	        * @param score
	        * @throws Exception
	        * @Description:
	 */
	private void InsertLog(String rowKeyAbnormal, String rowValue,String score ,Collector<AbnStatisTuple> out ) throws Exception {
		//String row = dip + "*" + dport + "*" + sip + "*" + gisstr + "*" + type + "*" + idcno;
		
		
		
		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String flowDir = splits[6];
		String sdIP = dip + "*" + idcno + "*" + sip;

		String[] gisArray = gis.split("#",-1);
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];//市
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];
		String areaName = gisArray[6];//省
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		String evaluate = score;
//		String abnormal = "1";
		String delayTime = "";
		
		double maxValue =  streamOctetsMap.get(triple)/(1024*60);
		mapHighFlow.put(rowKeyAbnormal, maxValue);
		String desc = String.format("异常连接, 峰值流量[%f KB/s]", maxValue);
		
		long attnum = 0;
		String avgdelay = String.format("%d", delaytimeMap.get(sdIP)/delayTimePVMap.get(sdIP));
		delayTime = "平均延迟："+ avgdelay + "s;最大延时：" + maxDelayMap.get(sdIP) + "s;最小延时：" + minDelayMap.get(sdIP)+"s;";
		attnum = sdPVMap.get(sdIP);
		
		String accessTimeKey = dip + "*" + sip + "*" + destport;
	
		if(!abnRowKeyStart.containsKey(accessTimeKey))
			return;
		
		String dateStr =  abnRowKeyStart.get(accessTimeKey);
		
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		
		Date dtTime = sdf.parse(dateStr);
		
		String dateStr_h = sdf_h.format(dtTime);
		String dateStr_d = sdf_d.format(dtTime);
		String dateStr_hour = sdf_hour.format(dtTime);
		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		row.add(new DataColumn("ROWKEY", rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP", sip));
		row.add(new DataColumn("DESTPORT", destport));
		row.add(new DataColumn("ACCESSTIME", Long.parseLong(dateStr)));
		row.add(new DataColumn("ABRNORMAL", 1));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", attnum+""));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", areaName));
		row.add(new DataColumn("FLOWDIRECTION", flowDir));
		row.add(new DataColumn("UPSTREAMOCTETS", streamOctetsMap.get(triple)));
		row.add(new DataColumn("UPSTREAMPACKET", streamPacketMap.get(triple)));
		row.add(new DataColumn("ATTTYPE","1"));
		
//		System.out.println("AbnormalCheck#######Into MongoDB#####" + row.toString());
		importtool.InsertRowData(sAbnormalTable, row);	
		
		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("SESSIONMIDDLE");
		record.setAttackNum(1l);
		record.setUpStreamOctets(streamOctetsMap.get(triple));
		record.setUpStreamPacket(streamPacketMap.get(triple));
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(areaName);
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(dateStr);
		out.collect(record);
//		collector.emit(new Values(dip,sip,destport,"",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,areaName,
//				1l,dateStr_hour,"SESSIONMIDDLE",idcno,sourceAreaCountry,Long.valueOf(streamOctetsMap.get(triple)),Long.valueOf(streamPacketMap.get(triple))));
		memStreamOctetsMap.put(rowKeyAbnormal, streamOctetsMap.get(triple));
		memStreamPacketMap.put(rowKeyAbnormal, streamPacketMap.get(triple));

	}
	
	/**
	 * 已发现的异常连接，更新字节和数据包值
	        * @param rowKeyAbnormal
	        * @param row
	        * @throws Exception
	        * @Description:
	 */
	private void UpdateLog(String rowKeyAbnormal, String rowValue,Collector<AbnStatisTuple> out ) throws Exception {
		
		String triple = rowValue;
		String[] splits = triple.split("\\*",-1);
		String destport = splits[1];
		String gis = splits[3];
		String protoType = splits[4];
		String sip = splits[2];
		String dip = splits[0];
		String idcno = splits[5];
		String sdIP = dip + "*" + idcno + "*" + sip;
		
		String accessTimeKey = dip + "*" + sip + "*" + destport;
		
		if(!abnRowKeyStart.containsKey(accessTimeKey))
			return;
		
		String dateStr =  abnRowKeyStart.get(accessTimeKey);
		
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");  
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		
		Date dtTime = sdf.parse(dateStr);
		
		String dateStr_h = sdf_h.format(dtTime);
		String dateStr_d = sdf_d.format(dtTime);
		String dateStr_hour = sdf_hour.format(dtTime);
		

		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];//市
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];
		String areaName = gisArray[6];//省
		
		
		if(memStreamOctetsMap.containsKey(rowKeyAbnormal)) {
			memStreamOctetsMap.put(rowKeyAbnormal, memStreamOctetsMap.get(rowKeyAbnormal) 
					+ streamOctetsMap.get(triple));
		}
		
		if(memStreamPacketMap.containsKey(rowKeyAbnormal)) {
			memStreamPacketMap.put(rowKeyAbnormal, memStreamPacketMap.get(rowKeyAbnormal) 
					+ streamPacketMap.get(triple));
		}
		
		double maxValue = (double)streamOctetsMap.get(triple)/(double)60/(double)1024 ;
		
		if(mapHighFlow.containsKey(rowKeyAbnormal))
		{
			double maxValueMem = mapHighFlow.get(rowKeyAbnormal);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(rowKeyAbnormal, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(rowKeyAbnormal, maxValue);
		}
		String desc = String.format("异常连接, 峰值流量[%f KB/s]", maxValue);
		
		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("UPSTREAMOCTETS", memStreamOctetsMap.get(rowKeyAbnormal)));
		row.add(new DataColumn("UPSTREAMPACKET", memStreamPacketMap.get(rowKeyAbnormal)));
		row.add(new DataColumn("DESC", desc));
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(sAbnormalTable, filter, row);
		
		AbnStatisTuple record = new AbnStatisTuple();
		record.setDestIP(dip);
		record.setSourceIP(sip);
		record.setType("SESSIONMIDDLE");
		record.setUpStreamOctets(streamOctetsMap.get(triple));
		record.setUpStreamPacket(streamPacketMap.get(triple));
		record.setSourceCountry(sourceAreaCountry);
		record.setSourceProvince(areaName);
		record.setAttackType("1");
		record.setAttackMax(maxValue);
		record.setAttTime(dateStr);
		out.collect(record);
		
//		collector.emit(new Values(dip,sip,destport,"",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,areaName,
//				0l,dateStr_hour,"SESSIONMIDDLE",idcno,sourceAreaCountry,Long.valueOf(streamOctetsMap.get(triple)),Long.valueOf(streamPacketMap.get(triple))));
		
		
	}
	
	/**
	 * 判断连接消失后，更新结束时间
	        * @param rowKeyAbnormal
	        * @Description:
	 */
	private void EndLog(String rowKeyAbnormal) {
				
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmmss");
		String endtime = sdf.format(abnRowKeyUpdate.get(rowKeyAbnormal) + 60*1000L);
		
		String sAbnormalTable = "SDS_ABNORMAL_LOG2";
		MongoImportTool importtool = MongoImportTool.getInstance();
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ENDTIME", Long.parseLong(endtime)));
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(sAbnormalTable, filter, row);
	
		//清理作案现场
		memStreamOctetsMap.remove(rowKeyAbnormal);
		memStreamPacketMap.remove(rowKeyAbnormal);
		
	}
	
	/**
	 * 
	        * @param abnNum 每分钟累积包数
	        * @return
	        * @Description:
	 */
	private String Score(long abnNum) {
		String score = "100";
		if(abnNum <= 50 ) {
			score = "100";
		} else if( abnNum > 50 && abnNum <= 100) {
			score = "60";
		} else if( abnNum > 200) {
			score = "40";
		}
		return score;
	}
	
	/**
	 * 
	  *  检查各个内存中MAP的大小，为清理内存提供依据
	  * @Description:
	 */
	private void CheckMapSize() {
		Logger.getLogger(AbnormalCheckImport.class).info("##HM abnStartTime=" + abnStartTime.size());
		Logger.getLogger(AbnormalCheckImport.class).info("##HM abnConfirm=" + abnUpdateTime.size());
		Logger.getLogger(AbnormalCheckImport.class).info("##HM abnDIPStatus=" + abnDIPStatus.size());
		Logger.getLogger(AbnormalCheckImport.class).info("##HM abnRowKeyUpdate=" + abnRowKeyUpdate.size());
		Logger.getLogger(AbnormalCheckImport.class).info("##HM memStreamPacketMap=" + memStreamPacketMap.size());
		Logger.getLogger(AbnormalCheckImport.class).info("##HM memStreamOctetsMap=" + memStreamOctetsMap.size());
		
	}


}
